import pandas as pd
import os
import sys
import time
import numpy as np
import datetime
import threading
from clickhouse_driver import Client

#财务数据

lock = threading.Lock()


#全局
mylist = []
# 多线程并发执行
def process_holder_merge_all_concurrency(path, save_to):
    if os.path.exists(save_to):
        print("文件已经存在，跳过！")
        return

    start = time.time()
    pd.set_option('display.unicode.ambiguous_as_wide', True)  # 处理数据的列标题与数据无法对齐的情况
    pd.set_option('display.unicode.east_asian_width', True)  # 无法对齐主要是因为列标题是中文
    pd.set_option('display.max_columns', None)

    list = os.listdir(path)
    for i in range(0, len(list)):
        # 指定字段类型，否则000删除了
        if os.path.isfile(os.path.join(path, list[i])):
            # 多线程执行
            t = threading.Thread(name='t_' + list[i], target=merge_csv, args=(os.path.join(path, list[i]),))
            t.start()
            t.join()  # 阻塞，直到执行完毕，才会执行主线程
        else:
            print("跳过文件夹：%s" % (list[i]))

    print("开始合并...")
    print(mylist)
    df = pd.DataFrame(mylist, columns={'采集日期': str, '股票代码': str, '报告期': str,
                                       '股东数量': int, '股价(不用)': str, '': str})
    print(df.head())
    #删除多余列
    df = df.drop(columns=['股价(不用)',''])
    #处理 -
    # df = df.replace('-', '0')
    print(df.head())
    # 保存的时候000被删除了,RE:文件中没有删除000，但是excel打开时看不到000，
    df.to_csv(save_to, encoding="utf_8_sig", index=False)  # index设置为False，就不会有第一列的序号
    end = time.time()
    print("concurrency time: %s" % (end - start))

def merge_csv(path):
    append = pd.read_csv(path, dtype={'股票代码': str}, encoding="UTF-8",
                         index_col=False)  # 取消索引列，否则错位
    print("线程：%s 分析文件： %s" % (threading.current_thread().name, path))
    size = len(append)
    append = append.iloc[0:size, :]  # 行，列
    # 需要加锁
    lock.acquire()
    for item in append.values.tolist():
        mylist.append(item)
    lock.release()


#离线批处理，单个文件（合并后的文件）
def batch_process(path):
    client = Client(host="192.168.50.40", user="default", password="9defbcg", port="9090", database="ods")
    print("start batch process holder data...")
    #读取csv数据，必须指定类型，否则会推测为int，导致异常
    #
    df = pd.read_csv(path, dtype={'采集日期': str, '股票代码': str, '报告期': str,
                                       '股东数量': int}
                     , encoding="UTF-8", index_col=False)  # 取消索引列，否则错位

    # df['采集日期'] = '20230818'

    client.execute('INSERT INTO ods_stock_holder_jka VALUES', df.values.tolist())
    #关闭连接 TODO
    print("batch process stock holder data finished!")

# 必须有main方法，调度才能执行
def main():
    print("holder_etl start....")

    if len(sys.argv) > 1:
        path = sys.argv[1]
        save_to = sys.argv[2]
    # path = "G:\\task-pipeline\\df_holder\\20230821"
    # save_to = "G:\\task-pipeline\\df_holder\\all_20230821.csv"

    # 合并资金流入数据 耗时：30s
    # process_finance_merge_all_concurrency("G:\\task-pipeline\\df_finance\\20230819\\test", "G:\\task-pipeline\\df_finance\\20230819\\test\\test.csv")
    process_holder_merge_all_concurrency(path, save_to)
    #批量导入到CK
    batch_process(save_to)


if __name__ == '__main__':
    main()



